package com.tangji.signaltower.disruptor;

import com.jsoniter.output.JsonStream;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.tangji.signaltower.bean.MarketDataEntity;
import com.tangji.signaltower.task.MarketDataVo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created with IntelliJ IDEA.
 * User: yujianfu (mr.vencnet@gmail.com)
 * Date: 2020-07-03
 * Time: 14:42
 */
public class DisruptorHandler {
    private static final Logger logger = LoggerFactory.getLogger(DisruptorHandler.class);
    public static Disruptor<MarketDataVo> disruptor = null;

    public static void init() {
        getDisruptor();
    }

    //创建disruptor
    public static Disruptor<MarketDataVo> getDisruptor() {

        if (disruptor != null) {
            return disruptor;
        }

        synchronized (DisruptorHandler.class) {
            // RingBuffer生产工厂,初始化RingBuffer的时候使用
            EventFactory<MarketDataVo> factory = new EventFactory<MarketDataVo>() {
                @Override
                public MarketDataVo newInstance() {
                    return new MarketDataVo();
                }
            };

            // 阻塞策略
            SleepingWaitStrategy strategy = new SleepingWaitStrategy();

            // 指定RingBuffer的大小
            int bufferSize = 1024 * 1024;

            ExecutorService executor = Executors.newCachedThreadPool();

            // 创建disruptor，采用单生产者模式
            disruptor = new Disruptor<MarketDataVo>(factory, bufferSize, executor, ProducerType.SINGLE, strategy);

            // 设置EventHandler
            disruptor.handleEventsWith(new TradeTransactionInDBHandler());

            // 启动disruptor的线程
            disruptor.start();
        }

        return disruptor;
    }

    public static void publish(MarketDataVo marketDataEntity) {
        logger.info("发布者发布消息,marketDataEntity:{}", JsonStream.serialize(marketDataEntity));
        //发布事件
        RingBuffer<MarketDataVo> ringBuffer = disruptor.getRingBuffer();
        // 获取下一个可用位置的下标
        long sequence = ringBuffer.next();
        try {
            // 返回可用位置的元素
            MarketDataVo event = ringBuffer.get(sequence);
            event.setBusinessTime(marketDataEntity.getBusinessTime())
                    .setMarketPrice(marketDataEntity.getMarketPrice())
                    .setMarketTimestamp(marketDataEntity.getMarketTimestamp())
                    .setType(marketDataEntity.getType())
                    .setRealType(marketDataEntity.getRealType())
                    .setHistoryData(marketDataEntity.getHistoryData());
        } finally {
            ringBuffer.publish(sequence);
        }
    }

}
